package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.UserJumpStats;
import com.atguigu.utils.MyClickHouseUtil;
import com.atguigu.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.PatternTimeoutFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.List;
import java.util.Map;

//数据流： web/app -> Nginx -> 日志服务器(xx.log) -> Flume -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp -> ClickHouse
//程  序： Mock -> Flume(f1.sh) -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK) -> UserJumpVisitDim10sApp -> ClickHouse(ZK)
public class UserJumpVisitDim10sApp {

    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);  //生产环境设置为Kafka分区数

        //        //1.1 开启CheckPoint
        //        env.enableCheckpointing(5 * 60000L);
        //        env.getCheckpointConfig().setCheckpointTimeout(5 * 60000L);
        //        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000L);
        //        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,5000L));
        //
        //        //1.2 指定状态后端
        //        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/xxxx/xxx"));

        //TODO 2.读取Kafka dwd_page_log 主题数据创建流
        String sourceTopic = "dwd_page_log";
        String groupId = "user_jump_visit_dim_10s_app_210927";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(sourceTopic, groupId));

        //TODO 3.将每行数据转换为JSON对象并提取时间戳生成Watermark
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaDS.map(JSON::parseObject);
        SingleOutputStreamOperator<JSONObject> jsonObjWithWMDS = jsonObjDS.assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(12)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getLong("ts");
            }
        }));

        //TODO 4.按照Mid分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjWithWMDS.keyBy(json -> json.getJSONObject("common").getString("mid"));

        //TODO 5.定义模式序列
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) {
                String lastPageId = value.getJSONObject("page").getString("last_page_id");
                return lastPageId == null;
            }
        }).next("next").where(new SimpleCondition<JSONObject>() {
            @Override
            public boolean filter(JSONObject value) {
                String lastPageId = value.getJSONObject("page").getString("last_page_id");
                return lastPageId == null;
            }
        }).within(Time.seconds(10));

//        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start").where(new SimpleCondition<JSONObject>() {
//            @Override
//            public boolean filter(JSONObject value) {
//                String lastPageId = value.getJSONObject("page").getString("last_page_id");
//                return lastPageId == null;
//            }
//        })
//                .times(2)      //默认为宽松近邻  followedBy
//                .consecutive() // 指定循环模式为严格近邻  next
//                .within(Time.seconds(10));

        //TODO 6.将模式序列作用到流上
        PatternStream<JSONObject> patternStream = CEP.pattern(keyedStream, pattern);

        //TODO 7.提取匹配上的事件以及超时事件，并Union
        OutputTag<JSONObject> timeOutTag = new OutputTag<JSONObject>("timeOut") {
        };
        SingleOutputStreamOperator<JSONObject> selectDS = patternStream.select(timeOutTag,
                new PatternTimeoutFunction<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject timeout(Map<String, List<JSONObject>> map, long l) throws Exception {
                        return map.get("start").get(0);
                    }
                }, new PatternSelectFunction<JSONObject, JSONObject>() {
                    @Override
                    public JSONObject select(Map<String, List<JSONObject>> map) throws Exception {
                        return map.get("start").get(0);
                    }
                });
        DataStream<JSONObject> timeOutDS = selectDS.getSideOutput(timeOutTag);

        selectDS.print("SelectDS>>>>>>>>>>>");
        timeOutDS.print("TimeOutDS>>>>>>>>>>>");

        DataStream<JSONObject> unionDS = selectDS.union(timeOutDS);

        //TODO 8.将数据转换为JavaBean对象
        SingleOutputStreamOperator<UserJumpStats> userJumpStatsDS = unionDS.map(line -> {
            JSONObject common = line.getJSONObject("common");
            return new UserJumpStats("", "",
                    common.getString("vc"),
                    common.getString("ch"),
                    common.getString("ar"),
                    common.getString("is_new"),
                    1L,
                    line.getLong("ts")
            );
        });

        //TODO 9.分组、开窗、聚合
        //重新给定时间戳生成Watermark
        SingleOutputStreamOperator<UserJumpStats> userJumpStatsWithNewWMDS = userJumpStatsDS.assignTimestampsAndWatermarks(WatermarkStrategy.<UserJumpStats>forBoundedOutOfOrderness(Duration.ofSeconds(12)).withTimestampAssigner(new SerializableTimestampAssigner<UserJumpStats>() {
            @Override
            public long extractTimestamp(UserJumpStats element, long recordTimestamp) {
                return element.getTs();
            }
        }));
        SingleOutputStreamOperator<UserJumpStats> result = userJumpStatsWithNewWMDS.keyBy(new KeySelector<UserJumpStats, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(UserJumpStats value) throws Exception {
                return new Tuple4<>(value.getAr(),
                        value.getCh(),
                        value.getIs_new(),
                        value.getVc());
            }
        })
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<UserJumpStats>() {
                    @Override
                    public UserJumpStats reduce(UserJumpStats value1, UserJumpStats value2) throws Exception {
                        value1.setUj_ct(value1.getUj_ct() + value2.getUj_ct());
                        return value1;
                    }
                }, new WindowFunction<UserJumpStats, UserJumpStats, Tuple4<String, String, String, String>, TimeWindow>() {
                    @Override
                    public void apply(Tuple4<String, String, String, String> key, TimeWindow window, Iterable<UserJumpStats> input, Collector<UserJumpStats> out) throws Exception {

                        //获取数据
                        UserJumpStats userJumpStats = input.iterator().next();

                        //设置窗口信息
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        userJumpStats.setStt(sdf.format(window.getStart()));
                        userJumpStats.setEdt(sdf.format(window.getEnd()));

                        //输出数据
                        out.collect(userJumpStats);
                    }
                });

        //TODO 10.将数据输出到ClickHouse
        result.print("result>>>>>>>>>>>>>>");
        result.addSink(MyClickHouseUtil.getClickHouseSink("insert into dws_userjump_vc_ch_isnew_ar_10s values(?,?,?,?,?,?,?,?)"));

        //TODO 11.启动任务
        env.execute("UserJumpVisitDim10sApp");

    }

}
